package eg;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.filecache.DistributedCache;

public class StringMatching {
  public static class Map extends Mapper<LongWritable, Text, WordKey, Text> {
     
     private Text location = new Text();
     private File stringListFile;
     private Path[] filepath;
     

     @Override
     public void setup(Context context) throws IOException, InterruptedException{
    	   
    	    filepath = DistributedCache.getLocalCacheFiles(context.getConfiguration());
    	  //stringListFile = new File(filepath[0].toUri()); 	//normal case
    	    stringListFile = new File("./query.dat"); 			//fix URI is not absolute
    	}
     
     
     public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  	 
      
         String currentFile = ((FileSplit) context.getInputSplit()).getPath().getName();
    	 BufferedReader linereader = new BufferedReader(new InputStreamReader(new FileInputStream(stringListFile)));
    	 
    	 String KeyWord;
    	 int Loc =0;
    	 boolean matched;
    	 int keyOffset = (int)key.get();
			
    	 byte[] longByte = value.getBytes();
    	 
    	 while((KeyWord = linereader.readLine())!=null){
				
				Loc = 0;
				
				byte[] keyByte = new byte[KeyWord.length()];
				keyByte = KeyWord.getBytes();
				String word = currentFile+","+KeyWord;
				
	    		for(;Loc<((longByte.length)-1-(int)longByte[0]);Loc++){ 
	    			
	    			if(longByte[Loc+1]==keyByte[0]){
	    				matched = false;
	    				
			    		for(int i=1;i<keyByte.length;i++){
			    			if(longByte[Loc+1+i]==keyByte[i]){
			    				matched = true;
			    			}else{
			    				matched = false;
			    				break;
			    			}
	    				}
			    		if(matched){
			    			
			    			int realLoc = Loc+keyOffset;
				    		location.set(Integer.toString(realLoc));
				    		WordKey wordKey = new WordKey(word, new Long(realLoc));
			    			context.write(wordKey, location);
			    		}
	    			}
	    		}

	    	 }

    	 linereader.close();
    	
     }
  }

  public static class Reduce extends Reducer<WordKey, Text, Text, Text> {
		
	  
	  private Text word = new Text();
	  
	  public void reduce(WordKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		
    	 boolean firstKey = true;
    	 word.set(key.toString());
    	 for (Text value : values) {
   		  context.write(firstKey ? word : null, value);
   	    firstKey = false;
   	  }
    	 

     }
}
  
  public static void main(String[] args) throws Exception {
    
	 Configuration conf = new Configuration();  
	 
	//block size 64 MB
	 conf.set("mapred.max.split.size","67108864");
	 
     Job job = new Job(conf, "StringMatching");
     job.setJarByClass(StringMatching.class);
     
     job.setPartitionerClass(NaturalKeyPartitioner.class);
	 job.setGroupingComparatorClass(NaturalKeyGroupingComparator.class);
	 job.setSortComparatorClass(CompositeKeyComparator.class);
	 
	 job.setMapOutputKeyClass(WordKey.class);
	 job.setMapOutputValueClass(Text.class);
	 
     job.setOutputKeyClass(Text.class);
     job.setOutputValueClass(Text.class);
     	     
     job.setMapperClass(Map.class);
     job.setReducerClass(Reduce.class);
     
     job.setInputFormatClass(CustomFileInputFormat.class);
     job.setOutputFormatClass(CustomTextOutputFormat.class);
     
     DistributedCache.addCacheFile(new URI("./wordlist/query.dat"), job.getConfiguration());
     FileInputFormat.addInputPath(job, new Path(args[0]));
     FileOutputFormat.setOutputPath(job, new Path(args[1]));
     job.waitForCompletion(true);


  }
}